package ex;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;

public class PublicSentimentTopology {
    private static final String TOPOLOGY_NAME = "publicSentimentTopology";
    private static final String SPOUT_NAME = "spoutMock";
    private static final String BOLT_MATCHING = "boltMatching";
    private static final String BOLT_IMPORTANT_INFO = "importantInfo";
    private static final String BOLT_FOCUS_INFO = "focusInfo";
    private static final String BOLT_HOT_INFO = "hotInfo";
    public static final String STATE_IMPORTANT = "important";
    public static final String STATE_FOCUS = "focus";
    public static final String STATE_HOT = "hot";


    public static void main(String[] args) throws Exception {
        SpoutMock spoutMock = new SpoutMock();
        BoltMatching boltMatching = new BoltMatching();
        BoltImportantInfo importantInfo = new BoltImportantInfo();
        BoltFocusInfo focusInfo = new BoltFocusInfo();
        BoltHotInfo hotInfo = new BoltHotInfo();

        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout(SPOUT_NAME, spoutMock);
        topologyBuilder.setBolt(BOLT_MATCHING, boltMatching).shuffleGrouping(SPOUT_NAME);
        topologyBuilder.setBolt(BOLT_IMPORTANT_INFO, importantInfo).globalGrouping(BOLT_MATCHING, STATE_IMPORTANT);
        topologyBuilder.setBolt(BOLT_FOCUS_INFO, focusInfo).globalGrouping(BOLT_MATCHING, STATE_FOCUS);
        topologyBuilder.setBolt(BOLT_HOT_INFO, hotInfo).globalGrouping(BOLT_MATCHING, STATE_HOT);

        StormTopology topology = topologyBuilder.createTopology();
        LocalCluster localCluster = new LocalCluster();
        Config config = new Config();
        localCluster.submitTopology(TOPOLOGY_NAME, config, topology);
    }
}
